package com.lamda.zhekou.master.session;

import com.alibaba.fastjson.JSONObject;
import com.lamda.zhekou.ZheKouApplication;
import com.lamda.zhekou.master.event.DataCallBack;
import com.lamda.zhekou.master.event.DataEvent;
import com.lamda.zhekou.master.event.DataListener;
import com.lamda.zhekou.master.event.Protocol;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;
import org.json.JSONException;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by MJiao on 15/8/25.
 */
@Component
public class WorkerSessionScheduler {

    private static Logger LOGGER = Logger.getLogger(WorkerSessionScheduler.class);

    public static final String SERVER_TO_CLIENT_REQ_ID_FLAG = "serverReqId";

    /**
     * 请求编号
     */
    private AtomicInteger reqId;
    /**
     * 比如服务器向客户端发送关机指令，服务器需要知道关机指令的执行结果
     * 请求编号<->回调函数 (服务器可能像客户端发送request请求，需要客户端的响应)
     */
    private Map<Integer, DataCallBack> cbs;

    /**
     * 监听器
     */
    private Map<String, List<DataListener>> listeners;

    /**
     *
     */
    private Set<WorkerSession> workerSessionList;

    public WorkerSessionScheduler() {
        listeners = new ConcurrentHashMap<>();
        cbs = new ConcurrentHashMap<>();
        workerSessionList = Collections.synchronizedSet(new HashSet<WorkerSession>());
        this.reqId = new AtomicInteger();
    }

    private void addWorkerSession(WorkerSession workerSession) {
        LOGGER.info("addWorkerSession::  当前的WorkerSessioin总数为:" + this.workerSessionList.size());
        for (WorkerSession session : workerSessionList) {
            try {
                if(session.getChannel() == null){
                    this.workerSessionList.remove(session);
                    continue;
                }
               closeOldWorkerSession(workerSession, session);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        LOGGER.info("addWorkerSession::  清理后WorkerSessioin总数为:" + this.workerSessionList.size());
        this.workerSessionList.add(workerSession);
        LOGGER.info("addWorkerSession::  添加后WorkerSessioin总数为:" + this.workerSessionList.size());
        workerSession.getChannel().closeFuture().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                WorkerSessionScheduler.this.workerSessionList.remove(workerSession);
            }
        });

    }

    private void closeOldWorkerSession(WorkerSession workerSession, WorkerSession oldWorkerSession) throws InterruptedException {
        if(workerSession.getWorkerIp().equals(oldWorkerSession.getWorkerIp())){
            if(oldWorkerSession.getChannel() == null){
                this.workerSessionList.remove(oldWorkerSession);
                return;
            }
            // 开始的倒数锁
            final CountDownLatch countDownLatch = new CountDownLatch(1);
            oldWorkerSession.getChannel().close().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    LOGGER.info("关闭旧的WorkerSession:" + future.isSuccess());
                    // 每个选手到达终点时，end就减一
                    countDownLatch.countDown();
                }
            });
            countDownLatch.await(60, TimeUnit.SECONDS);
        }
    }

    public void checkOrRegisterWorkSession(Channel channel, String workerIp) {
        for (WorkerSession workerSession : workerSessionList) {
            if (workerSession.getChannel().equals(channel)) {
                return;
            }
        }
        WorkerSession workerSession = new WorkerSession();
        workerSession.setChannel(channel);
        workerSession.setUuid(UUID.randomUUID().toString());
        workerSession.setWorkerIp(workerIp);
        this.addWorkerSession(workerSession);
    }

    /**
     * Add event listener and wait for broadcast message.
     *
     * @param route
     * @param listener
     */
    public void on(String route, DataListener listener) {
        List<DataListener> list = listeners.get(route);
        if (list == null) {
            list = new ArrayList<DataListener>();
        }
        list.add(listener);
        listeners.put(route, list);
    }

    /**
     * Touch off the event and call listeners corresponding route.
     *
     * @param route
     * @param message
     * @param channel
     * @return true if call success, false if there is no listeners for this
     * route.
     */
    public void emit(String route, JSONObject message, Channel channel) {
        if(StringUtils.isEmpty(route)){
            return;
        }
        List<DataListener> list = listeners.get(route);
        if (list == null) {
            LOGGER.warn("there is no listeners for route:[" + route + "]");
            return;
        }
        for (DataListener listener : list) {
            DataEvent event = new DataEvent(this, message);
            listener.receiveData(event, channel);
        }
    }

    /**
     * Master向Worker发送请求
     *
     * @param route
     * @param jsonObject
     * @param dataCallBack
     */
    public void push(String route, JSONObject jsonObject, DataCallBack dataCallBack) {
        String workerIp = jsonObject.getString("workerIp");
        this.requestWithArgs(workerIp, route, jsonObject, dataCallBack);
    }

    /**
     * 通知客户端不需要响应
     *
     * @param route
     * @param jsonObject
     * @param dataCallBack
     */
    public void notify(String route, JSONObject jsonObject, DataCallBack dataCallBack) {
        String workerIp = jsonObject.getString("workerIp");
        this.requestWithArgs(workerIp, route, jsonObject, dataCallBack);
    }

    /**
     * Client send requestWithArgs to the server and get response data.
     *
     * @param workerIp
     * @param route
     * @param msg
     * @param dataCallBack
     */
    private void requestWithArgs(String workerIp, String route, JSONObject msg, DataCallBack dataCallBack) {

        msg = filter(msg);

        //当currReqId=-1时 不需要客户端向服务器返送返回值
        int currReqId = -1;
        if (dataCallBack != null) {
            //对reqId进行益处容错
            if (this.reqId.get() > 999999996) {
                this.reqId.set(0);
            }
            //reqId自增
            currReqId = this.reqId.incrementAndGet();
            cbs.put(currReqId, dataCallBack);
        }

        sendMessage(workerIp, currReqId, route, msg);
    }


    /**
     * Add timestamp to message.
     *
     * @param msg
     * @return msg
     */
    private JSONObject filter(JSONObject msg) {
        if (msg == null) {
            msg = new JSONObject();
        }
        long date = System.currentTimeMillis();
        try {
            msg.put("timestamp", date);
        } catch (JSONException e) {
            e.printStackTrace();
        }
        return msg;
    }

    /**
     * Send message to the server side.
     *
     * @param reqId requestWithArgs id
     * @param route requestWithArgs route
     * @param msg   push message
     */
    private void sendMessage(String workerIp, int reqId, String route, JSONObject msg) {

        String content = Protocol.encode(reqId, route, msg);
        if (reqId < 0) {
            content = Protocol.encode(route, msg);
        }

        LOGGER.info("向客户端发送命令:" + content);


        /*
        * 便利目标Channel
         */
        Channel channel = null;
        for (WorkerSession workerSession : workerSessionList) {
            if (workerIp.equals(workerSession.getWorkerIp())) {
                channel = workerSession.getChannel();
                break;
            }
        }
        /**
         * 如果目标channel不可用
         */
        DataCallBack dataCallBack = cbs.get(reqId);
        if ((channel == null || !channel.isWritable()) && dataCallBack != null) {
            LOGGER.info("目标channel不可能用:" + channel);
            org.json.JSONObject result = new org.json.JSONObject();
            result.put("route", msg.get("route"));
            result.put("succ", msg.get("false"));
            result.put("msg", "vps不存在或者脱离了Master的管理,请稍后再试");

            dataCallBack.responseData(msg);

            return;
        }
        //向客户端发送消息
        channel.writeAndFlush(content).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                LOGGER.info("消息是否push成功:" + future.isSuccess());
            }
        });

    }

    /**
     * 处理接收到的小心： 进行事件分发等操作
     *
     * @param jsonObject
     * @param ctx
     */
    public void handler(JSONObject jsonObject, ChannelHandlerContext ctx) {

        try {

            String route = jsonObject.getString("route");
            jsonObject.remove("route");
            //该请求是服务器发向客户端的：获取该请求Id,判断是否有请求形handler
            if (jsonObject.containsKey(WorkerSessionScheduler.SERVER_TO_CLIENT_REQ_ID_FLAG)) {
                int reqId = jsonObject.getIntValue(WorkerSessionScheduler.SERVER_TO_CLIENT_REQ_ID_FLAG);
                if (cbs.get(reqId) != null) {
                    DataCallBack dataCallBack = cbs.get(reqId);
                    dataCallBack.responseData(jsonObject);
                }
            }
            //监听器处理
            emit(route, jsonObject, ctx.channel());
        } catch (JSONException e) {
            jsonObject = new JSONObject();
            jsonObject.put("msg", e.getMessage());
            emit("error", jsonObject, ctx.channel());
        }
    }


}
